package us.luosl.lslt.asyn.pipeline;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 基于内存的管道
 * @param <T>
 */
public class MemoryPipeline<T> implements Pipeline<T> {

    private LinkedBlockingQueue<T> queue;

    public MemoryPipeline(int maxSize) {
        this.queue = new LinkedBlockingQueue<>(maxSize);
    }

    @Override
    public boolean push(T data) {
        return queue.offer(data);
    }

    @Override
    public List<T> pull(int size) {
        List<T> collector = new ArrayList<>();
        T item;
        do{
            item = queue.poll();
            if(null != item){
                collector.add(item);
            }
        }while(null != item && collector.size() < size);

        return collector;
    }

}
